<?php

/*
 * This file is part of the slince/dragbox package.
 *
 * (c) Slince <taosikai@yeah.net>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace DragBox\Server\Worker;

use DragBox\Common\Exception\BadRequestException;
use function Dragbox\Common\jsonBuffer;
use DragBox\Common\Protocol\Spike;
use DragBox\Server\Filesystem\FilesystemInterface;
use React\EventLoop\LoopInterface;
use React\Socket\ConnectionInterface;
use React\Socket\Server as Socket;
use React\Stream\WritableResourceStream;

class Worker implements WorkerInterface
{
    /**
     * @var FilesystemInterface
     */
    protected $filesystem;

    /**
     * @var Socket
     */
    protected $socket;

    /**
     * @var LoopInterface
     */
    protected $loop;

    /**
     * @var string
     */
    protected $address;

    public function __construct(
        LoopInterface $loop,
        FilesystemInterface $filesystem,
        $address
    ) {
        $this->loop = $loop;
        $this->filesystem = $filesystem;
        $this->address = $address;
    }

    /**
     * {@inheritdoc}
     */
    public function start()
    {
        $this->socket = new Socket($this->address, $this->loop);
        $this->socket->on('connection', [$this, 'handleConnection']);
    }

    public function handleConnection(ConnectionInterface $connection)
    {
        jsonBuffer($connection,function($messages) use ($connection){
            $message = Spike::fromArray(reset($messages));
            if ($message->getAction() === 'start_upload') {

                $medata = $message->getBody()['metadata'];

                $connection->removeAllListeners('data');

                try {
                    $this->handleUpload($connection, $medata);
                } catch (\Exception $exception) {
                    $message = new Spike('upload_end', [
                        'message' => $exception->getMessage()
                    ], [
                        'code' => $exception->getCode() ?: 400
                    ]);
                    $connection->write($message);

                    $connection->removeAllListeners('data');
                    //Continue waiting upload.
                    $this->handleConnection($connection);
                }
            } else {
                throw new BadRequestException('Expect action "start_upload"');
            }
        });
    }

    public function handleUpload(ConnectionInterface $connection, $metadata)
    {
        $resource = $this->filesystem->createStream($metadata['key'], 'a');
        $writeStream = new WritableResourceStream($resource, $this->loop);
        $length = 0;
        $connection->on('data', function($data) use ($writeStream, $metadata, &$length, $connection){

            $length += strlen($data);
            $writeStream->write($data);

            //If receive ok
            if ($length === $metadata['size']) {
                $connection->write(new Spike('upload_end', [
                    'message' => 'upload ok'
                ], ['code' => 200]));
                $connection->removeAllListeners('data');
                //Continue waiting upload.
                $this->handleConnection($connection);

            } elseif ($length > $metadata['size']) {
                throw new BadRequestException(sprintf('The file size is exceeded. expect "%d", got "%d"',
                    $metadata['size'],
                    $length
                ));
            }
        });
    }
}